<?php
/**
 * RabbitMQ服务类
 * 封装RabbitMQ的生产者和消费者功能
 */

namespace app\common\library;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;
use think\Config;
use think\Exception;

class RabbitMQ
{
    /**
     * @var AMQPStreamConnection
     */
    protected $connection;
    
    /**
     * @var \PhpAmqpLib\Channel\AMQPChannel
     */
    protected $channel;
    
    /**
     * 配置参数
     * @var array
     */
    protected $config = [];
    
    /**
     * 初始化RabbitMQ连接
     */
    public function __construct()
    {
        $this->config = Config::get('rabbitmq');
        
        // 确保配置已加载
        if (empty($this->config)) {
            throw new Exception('RabbitMQ配置未加载');
        }
        
        // 设置默认值
        $this->config = array_merge([
            'timeout' => 3,
            'heartbeat' => 60
        ], $this->config);
        
        try {
            $this->connection = new AMQPStreamConnection(
                $this->config['host'],
                $this->config['port'],
                $this->config['user'],
                $this->config['password'],
                $this->config['vhost'],
                false,  // insist
                'AMQPLAIN',  // login_method
                null,  // login_response
                'en_US',  // locale
                $this->config['timeout'],  // connection_timeout
                $this->config['heartbeat'],  // read_write_timeout
                null,  // context
                true,  // keepalive
                $this->config['heartbeat']  // heartbeat
            );
            
            $this->channel = $this->connection->channel();
            
            // 声明交换机
            $this->channel->exchange_declare(
                $this->config['exchange'],
                'direct',  // type
                false,  // passive
                true,  // durable
                false  // auto_delete
            );
            
            // 声明队列
            $this->channel->queue_declare(
                $this->config['queue'],
                false,  // passive
                true,  // durable
                false,  // exclusive
                false  // auto_delete
            );
            
            // 绑定队列到交换机
            $this->channel->queue_bind(
                $this->config['queue'],
                $this->config['exchange'],
                $this->config['routing_key']
            );
            
        } catch (\Exception $e) {
            throw new Exception('RabbitMQ connection failed: ' . $e->getMessage());
        }
    }
    
    /**
     * 发送消息
     * @param mixed $data 要发送的数据
     * @param array $properties 消息属性
     * @return bool
     */
    public function send($data, $properties = [])
    {
        try {
            $messageBody = json_encode($data, JSON_UNESCAPED_UNICODE);

            $defaultProperties = [
                'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
                'content_type' => 'application/json',
                'timestamp' => time(),
            ];
            
            $properties = array_merge($defaultProperties, $properties);
            
            $message = new AMQPMessage($messageBody, $properties);

            $this->channel->basic_publish(
                $message,
                $this->config['exchange'],
                $this->config['routing_key']
            );
            
            return true;
        } catch (\Exception $e) {
            throw new Exception('RabbitMQ send message failed: ' . $e->getMessage());
        }
    }
    
    /**
     * 消费消息
     * @param callable $callback 回调函数
     * @param array $options 消费选项
     */
    public function consume(callable $callback, $options = [])
    {
        try {
            $defaultOptions = [
                'consumer_tag' => 'consumer_' . uniqid(),
                'no_local' => false,
                'no_ack' => false,
                'exclusive' => false,
                'nowait' => false,
                'ticket' => null,
                'arguments' => new AMQPTable([
                    'x-priority' => 10,
                ]),
            ];
            
            $options = array_merge($defaultOptions, $options);
            
            $this->channel->basic_consume(
                $this->config['queue'],
                $options['consumer_tag'],
                $options['no_local'],
                $options['no_ack'],
                $options['exclusive'],
                $options['nowait'],
                function ($message) use ($callback) {
                    try {
                        $data = json_decode($message->body, true);
                        if (json_last_error() !== JSON_ERROR_NONE) {
                            throw new Exception('Invalid JSON data');
                        }
                        
                        // 调用回调函数处理消息
                        $result = call_user_func($callback, $data);
                        
                        // 如果回调返回true或没有返回值，则确认消息
                        if ($result === true || $result === null) {
                            $message->ack();
                            
                            // 发送确认响应
                            if (isset($data['reply_to'])) {
                                $response = new AMQPMessage(
                                    json_encode(['status' => 'success', 'message' => 'Message received successfully'], JSON_UNESCAPED_UNICODE),
                                    [
                                        'correlation_id' => $message->get('correlation_id'),
                                        'content_type' => 'application/json',
                                        'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
                                    ]
                                );
                                $this->channel->basic_publish($response, '', $data['reply_to']);
                            }
                        } else {
                            // 拒绝消息并重新排队
                            $message->nack(false, true);
                        }
                    } catch (\Exception $e) {
                        // 记录错误日志
                        \think\Log::error('RabbitMQ consume error: ' . $e->getMessage());
                        
                        // 拒绝消息并丢弃
                        $message->nack(false, false);
                        
                        // 发送错误响应
                        if (isset($data['reply_to'])) {
                            $response = new AMQPMessage(
                                json_encode(['status' => 'error', 'message' => $e->getMessage()], JSON_UNESCAPED_UNICODE),
                                [
                                    'correlation_id' => $message->get('correlation_id'),
                                    'content_type' => 'application/json',
                                    'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
                                ]
                            );
                            $this->channel->basic_publish($response, '', $data['reply_to']);
                        }
                    }
                },
                $options['ticket'],
                $options['arguments']
            );
            
            // 持续处理消息
            while ($this->channel->is_consuming()) {
                $this->channel->wait();
            }
            
        } catch (\Exception $e) {
            throw new Exception('RabbitMQ consume failed: ' . $e->getMessage());
        }
    }
    
    /**
     * 关闭连接
     */
    public function close()
    {
        try {
            if ($this->channel) {
                $this->channel->close();
            }
            
            if ($this->connection) {
                $this->connection->close();
            }
        } catch (\Exception $e) {
            // 忽略关闭连接时的错误
        }
    }
    
    /**
     * 析构函数，确保连接被关闭
     */
    public function __destruct()
    {
        $this->close();
    }
}